package com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.k8s;

import com.google.common.collect.Maps;
import com.ds.infrastructure.hyperspace.common.ResolverTypeEnum;
import com.ds.infrastructure.hyperspace.common.config.K8sConfig;
import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.BaseResolverProvider;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.util.HttpRequestUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.ScheduledFuture;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.net.URI;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.k8s.RequestType.QUERY;
import static com.ds.infrastructure.hyperspace.container.endpoint.destination.resolver.k8s.RequestType.WATCH;
import static java.util.concurrent.TimeUnit.MILLISECONDS;


/**
 * @Author: yaozh
 * @Description:
 */
@Slf4j
public class K8sResolverProvider extends BaseResolverProvider {
    protected static final long READER_IDLE_TIME = 30L * 1000L;
    protected static final long WRITER_IDLE_TIME = 5L * 1000L;
    private static final String QUERY_URL_TEMPLATE = "%s://%s:%d/api/v1/namespaces/%s/endpoints/%s";
    private static final String WATCH_URL_TEMPLATE = "%s://%s:%d/api/v1/watch/namespaces/%s/endpoints/%s";
    private static final int MAX_FAST_CONNECTION_COUNT = 10;
    @Getter
    private final K8sConfig k8sConfig;
    @Getter
    private final K8sTokenStorage k8sTokenStorage;
    private final EventLoopGroup workerGroup;
    protected boolean parseSuccess = true;
    private int connectionFailCount = 0;
    private volatile ScheduledFuture<?> scheduledFuture;
    private URI queryUri;
    private URI watchUri;

    public K8sResolverProvider(NettyContainer nettyContainer, TargetNodeConfig targetNodeConfig, String namespace) {
        super(nettyContainer, targetNodeConfig, namespace);
        init(targetNodeConfig);
        this.k8sConfig = nettyContainer.getK8sConfig();
        this.k8sTokenStorage = nettyContainer.getK8sTokenStorage();
        this.workerGroup = nettyContainer.getResolverEventLoopGroup();
        scheduledFuture = workerGroup.schedule(this::query, 0, MILLISECONDS);
    }

    private void init(TargetNodeConfig targetNode) {
        String[] pathSplits = StringUtils.split(targetNode.getServicePath(), "/");
        if (pathSplits == null || pathSplits.length != 2) {
            String errorMsg = String.format("Target node service path invalid ! resolver info(host:%s) ,service path:%s"
                    , targetNode.getHost(), targetNode.getServicePath());
            throw new IllegalArgumentException(errorMsg);
        }
        String queryUrl = String.format(QUERY_URL_TEMPLATE,
                getProtocol(targetNode.getResolverType()), targetNode.getHost(), targetNode.getPort(), pathSplits[0], pathSplits[1]);
        queryUri = HttpRequestUtil.createUri(queryUrl);
        String watchUrl = String.format(WATCH_URL_TEMPLATE,
                getProtocol(targetNode.getResolverType()), targetNode.getHost(), targetNode.getPort(), pathSplits[0], pathSplits[1]);
        watchUri = HttpRequestUtil.createUri(watchUrl);
    }

    private String getProtocol(ResolverTypeEnum resolverType) {
        if (ResolverTypeEnum.K8S.equals(resolverType)) {
            return "http";
        }
        if (ResolverTypeEnum.K8SS.equals(resolverType)) {
            return "https";
        }
        throw new UnsupportedOperationException(
                "Only resolver type k8s or k8ss is supported, but current resolver type =" + resolverType);
    }

    public void query() {
        if (log.isDebugEnabled()) {
            log.debug("K8s query type request,uri:{}", queryUri);
        }
        Bootstrap b = getBootstrap();
        b.handler(new K8sResolverChannelInitializer(this, HttpRequestUtil.getSslContext(queryUri.getScheme()), QUERY, queryUri));
        b.connect(queryUri.getHost(), HttpRequestUtil.getPort(queryUri)).addListener(new ConnectFutureListener(this, queryUri));
    }

    private Bootstrap getBootstrap() {
        Bootstrap b = new Bootstrap();
        b.group(workerGroup);
        b.channel(NioSocketChannel.class);
        b.option(ChannelOption.TCP_NODELAY, true);
        b.option(ChannelOption.SO_REUSEADDR, true);
        b.option(ChannelOption.SO_KEEPALIVE, true);
        b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000);
        return b;
    }

    public void watch() {
        if (log.isDebugEnabled()) {
            log.debug("K8s watch type request,uri:{}", watchUri);
        }
        Bootstrap b = getBootstrap();
        b.handler(new K8sResolverChannelInitializer(this, HttpRequestUtil.getSslContext(watchUri.getScheme()), WATCH, watchUri));
        b.connect(watchUri.getHost(), HttpRequestUtil.getPort(watchUri)).addListener(new ConnectFutureListener(this, watchUri));
    }

    @Override
    public void cancel() {
        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
        }
    }

    public void reconnection() {
        if(isClosed()){
            return;
        }
        scheduledFuture = workerGroup.schedule(() -> {
            try {
                query();
            } catch (Exception e) {
                log.error("K8s resolver connection fail!", e);
            }

        }, getIntervalTime(), TimeUnit.SECONDS);
    }

    private Long getIntervalTime() {
        if (connectionFailCount > MAX_FAST_CONNECTION_COUNT) {
            return 3L;
        }
        if (!parseSuccess) {
            return 3L;
        }
        return 0L;
    }


    private static class ConnectFutureListener implements GenericFutureListener<ChannelFuture> {

        private final K8sResolverProvider resolverProvider;
        private final URI uri;

        public ConnectFutureListener(K8sResolverProvider resolverProvider, URI uri) {
            this.resolverProvider = resolverProvider;
            this.uri = uri;
        }

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!NettyContainer.STATUS.get()||resolverProvider.isClosed()) {
                return;
            }
            Channel newChannel = future.channel();
            if (!future.isSuccess()) {
                log.warn("K8s resolver connection fail! url:{},  fail count:{}", uri, resolverProvider.connectionFailCount);
                resolverProvider.connectionFailCount++;
                if (newChannel != null && newChannel.isOpen()) {
                    newChannel.flush().close();
                }
                resolverProvider.reconnection();
                return;
            }
            //reset connection fail count
            resolverProvider.connectionFailCount = 0;
            Map<CharSequence, Object> header = Maps.newHashMapWithExpectedSize(2);
            header.put(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            header.put(HttpHeaderNames.CONTENT_TYPE.toString(), HttpHeaderValues.APPLICATION_JSON);
            if (HttpScheme.HTTPS.name().contentEqualsIgnoreCase(uri.getScheme())) {
                setAuthorizationHeader(header, uri);
            }
            DefaultFullHttpRequest request =
                    HttpRequestUtil.createFullHttpRequest(uri, HttpVersion.HTTP_1_1, HttpMethod.GET, header);
            newChannel.writeAndFlush(request);
        }

        private void setAuthorizationHeader(Map<CharSequence, Object> header, URI uri) {
            String token = resolverProvider.k8sTokenStorage.getToken(uri.getHost());
            if (StringUtils.isBlank(token)) {
                return;
            }
            header.put(HttpHeaderNames.AUTHORIZATION, "Bearer " + token);
        }
    }


}
